package ranks.tasks.calc;

import java.io.IOException;

import mapred.example.base.TextMapper;
import mapred.util.FeatureMap;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

import ranks.Customize;
import ranks.Customize.DataType;
import ranks.base.Const;
import ranks.base.Data;

public class Calc_Mapper extends TextMapper {

    private static Data z;

    @Override
    public void map(Text key, Text value, OutputCollector<Text, Text> output,
            Reporter reporter) throws IOException {
        String text = value.toString();
        int pos = text.indexOf("\t");
        Data rank = new Data(text.substring(0, pos));
        output.collect(key, new Text(Const.OLD + rank));

        rank.add(z);
        String link = text.substring(pos + 1);
        output.collect(key, new Text(link));

        Long outDegree = rank.getValue(DataType.OUT_DEGREE);
        if (outDegree == 0) {
            output.collect(new Text("Z"), new Text(Const.RANK + rank));
        } else {
            FeatureMap map = FeatureMap.parse(link, Const.DELIM);
            for (String target : map.featureSet()) {
            	long weight = map.get(target);
            	Data send = Customize.send(key.toString(), rank, target, weight);
                output.collect(new Text(target), new Text(Const.RANK + send));
            }
        }
    }

    @Override
    public void setUp() {
        z = new Data(this.getJobConfConfig("Z"));
    }

}
